package cn._51doit.day05;


import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.GlobalWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.triggers.CountTrigger;
import org.apache.flink.streaming.api.windowing.triggers.PurgingTrigger;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;

/**
 * @create: 2021-10-20 11:18
 * @author: 今晚打脑斧
 * @program: CountWindowAllDemo
 * @Description:
 *
 * 不keyBy，直接划分滚动窗口，即按照ProcessingTime划分滚动窗口
 * 底层调用的是windowAll方法，传入的是ProcessingTime类型的窗口
 *
 * 调用的是windowAll方法得到是的nonKeyedWindow，window以及window function算子对应的DataStream并行度为1
 *
 * 我的理解就是,十秒钟进行一次数据的处理,不管数据来了多少条,十秒钟后才会处理这1秒钟内产生的数据
 * 然后再等十秒,对这十秒中内产生的新数据进行处理
 *
 **/
public class ProcessingTimeTumblingWindowAllDemo {
    public static void main(String[] args) throws Exception {
        Configuration configuration = new Configuration();
        configuration.setInteger("rest.port",8081);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(configuration);

        DataStreamSource<String> lines = env.socketTextStream("doit01", 8888);
        SingleOutputStreamOperator<Integer> nums = lines.map(Integer::parseInt);

        //滚动窗口,10秒滚动一次
        //AllWindowedStream<Integer, TimeWindow> windowedStream = nums.timeWindowAll(Time.seconds(10));
        AllWindowedStream<Integer, TimeWindow> windowedStream = nums.windowAll(TumblingProcessingTimeWindows.of(Time.seconds(10)));

        SingleOutputStreamOperator<Integer> res = windowedStream.sum(0);

        res.print();
        env.execute();

    }
}
